-module(partition_buffer).
-behaviour(gen_server).
-export([start_link/1, new_sender/2, sender/3, get_by_index/2]).
-export([init/1,handle_info/2,handle_call/3, handle_cast/2, terminate/2]).

start_link(ReducerList) ->
    gen_server:start({local, ?MODULE}, ?MODULE, [ReducerList], []).

init([ReducerList]) ->
    Len = length(lists:flatten(ReducerList)),
    SenderList = new_sender(Len, ReducerList),
    {ok, {SenderList,Len}}.

send_over2_all([]) ->
    ok;
send_over2_all(L) ->
    [H|T] = L,
    {_N, Pid} = H,
    Pid ! over,
    send_over2_all(T).

handle_call({over}, _, State) ->
    {SenderList, _Len} = State,
    send_over2_all(SenderList),
    {stop, normal, State};

handle_call({newinput, L}, _, State) ->
    {SenderList, Len} = State,
    helper_send(L, SenderList, Len),
    {reply, ok, State}.

helper_send([], _SenderList, _Len) ->
    ok;
helper_send(L, SenderList, Len) ->
    [{Key, Value}|T] = L,
    {_N, Sender_Pid} = get_by_index(SenderList, erlang:phash(Key, Len)),
    Sender_Pid ! {Key, Value},
    helper_send(T, SenderList, Len).
    
new_sender(0, _ReducerList) ->
    [];
new_sender(N, ReducerList) ->
    {N, Reducer_Pid} =get_by_index(ReducerList, N),
    Pid = spawn(partition_buffer, sender, [0, Reducer_Pid, []] ),
    [{N,Pid}|new_sender(N-1, ReducerList)].

sender(N, Reducer_Pid, L) when N > 100 ->
    Reducer_Pid ! L,
    sender(0, Reducer_Pid, []);
      
sender(N, Reducer_Pid, L) ->
    receive
	{Key, Value} ->
	    sender(N+1, Reducer_Pid, [{Key, Value}|L]);
	over ->
	    Reducer_Pid ! L,
	    Reducer_Pid ! {over, self()}
    end.

get_by_index([], _Index) ->
    false;
get_by_index([{N, Reducer_Pid}|_T], Index) when N == Index ->
    {N, Reducer_Pid};
get_by_index([_H|T], Index) ->
    get_by_index(T, Index).

	
handle_info(_Msg, _State) ->
    {reply, ok}.

handle_cast(_Msg, _State) ->
    {noreply, ok}.

terminate(_Args,_State) ->		
    ok.
